/*
 * MapReducer.c
 *
 *  Created on: 2011-8-14
 *      Author: ryan
 */
#include "MapReducer.h"
#include "Common.h"
#include "Step_One.h"
#include "Step_Two.h"
#include "Step_Three.h"

void splitTask(int argc, char **argv, TASK *task){
	gettimeofday(&alst,NULL);

	inputImageFileName[0] = '\0';
	refImageFileName[0] = '\0';
	outFilePath[0] = '\0';
	if (analizeOpt(argc, argv) < 3) {
		fprintf(stderr, "Pleas set the arg properly!\n");
		exit(0);
	}
	fileReading();

	int len = ref.width / MACHINES;
	int t = ref.width % MACHINES;
	if(t != 0){
		++len;
	}
	int start = 0;

	task->num = MACHINES;
	task->subtasks = malloc(sizeof(KEY_VALUE) * MACHINES);
	int k, i, size;
	for(k = 0; k < MACHINES; ++k){
		if(k == t && t != 0){
			--len;
		}
		KEY_VALUE *kv = &task->subtasks[k];
		kv->keySize = sizeof(int) * 3;
		kv->key = malloc(kv->keySize);
		int *p = (int*)kv->key;
		p[0] = k;
		p[1] = start;
		p[2] = len;
		start += len;

		kv->valueSize = sizeof(int) * 4 + sizeof(color) * (ref.width * ref.height + inp.width * inp.height);
		kv->value = malloc(kv->valueSize);
		unsigned char * q = (unsigned char *)kv->value;
		size = sizeof(int);
		memcpy(q, &ref.width, size);
		q += size;
		memcpy(q, &ref.height, size);
		q += size;
		memcpy(q, &inp.width, size);
		q += size;
		memcpy(q, &inp.height, size);
		q += size;
		size = sizeof(color)*ref.height;
		for(i = 0; i < ref.width; ++i){
			memcpy(q, ref.data[i], size);
			q += size;
		}
		size = sizeof(color) * inp.height;
		for (i = 0; i < inp.width; ++i) {
			memcpy(q, inp.data[i], size);
			q += size;
		}
	}

	gettimeofday(&aled,NULL);
	split_task_time = calculateTime(aled, alst);
	printf("Time used for spliting the task is : %lf\n", split_task_time);
	printf("Node |\timage_init\tnextcdp   \tbacktrace  \talg_finalize\toverhead\ttotal\n");
}

void map(KEY_VALUE *kin, KEY_VALUE *kout){
	gettimeofday(&alst,NULL);
	getImage(kin);
	mirror(&inp, &out);
	initialBuf(kin);
	gettimeofday(&aled,NULL);
	alg_image_init_time = calculateTime(aled, alst);

	gettimeofday(&alst,NULL);
	nextCDP();
	gettimeofday(&aled,NULL);
	nextcdp_time = calculateTime(aled, alst);

	gettimeofday(&alst,NULL);
	if(rank == MACHINES - 1){
		rankExtraction();
	}
	backTrace();
	if(rank == 0){
		extractPic();
	}
	clearBuf();
	gettimeofday(&aled,NULL);
	backtrace_time = calculateTime(aled, alst);

	gettimeofday(&alst,NULL);
	if (rank == 0) {
		getProjection(0);
		fileWriting();
	}
	finalize();
	gettimeofday(&aled,NULL);
	alg_finalize_time = calculateTime(aled, alst);
}

void reduce(KEY_VALUE *kin, KEY_VALUE *kout){

}
